/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package studio.raptor.ddal.dashboard.rule;


import com.google.common.base.Splitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.ddal.common.algorithm.AbstractHashRangeShardAlgorithm;
import studio.raptor.ddal.common.algorithm.ShardValue;
import studio.raptor.ddal.common.algorithm.SingleKeyShardAlgorithm;
import studio.raptor.ddal.common.exception.GenericException;
import studio.raptor.ddal.common.exception.code.RouteErrCodes;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 分片算法：分片字段为数字型，按区间路由
 * @author Sam
 * @since 3.0.0
 */
public class TestDateSingleKeyShardAlgorithm implements SingleKeyShardAlgorithm<Comparable<?>> {

    private static Logger logger = LoggerFactory.getLogger(TestDateSingleKeyShardAlgorithm.class);

    private static final String BASIC_DIR = "ddal/";
    private final String fileName;
    protected List<DateRange> ranges = new ArrayList<>();
    protected Map<String, String> shardTarget = new HashMap<>();
    protected String rangeText;

    public TestDateSingleKeyShardAlgorithm(String fileName) {
        try{
            this.fileName = fileName;
            readFile();
            parseRange();
            loadTargetMap();
        }catch (Exception e){
            throw new GenericException( RouteErrCodes.ROUTE_429, new Object[]{"处理时间分片路由算法配置文件:"+fileName+"错误"} );
        }
    }

    @Override
    public String doEqual(Collection<String> shards, ShardValue<Comparable<?>> shardValue) {
        Date value = (Date)shardValue.getValue();
        String suffix = SEPARATOR + shardTarget.get("" + rangeShardValue(value));
        for (String shard : shards) {
            if (shard.endsWith(suffix)) {
                return shard;
            }
        }
        throw new UnsupportedOperationException();
    }

    @Override
    public Collection<String> doIn(Collection<String> shardings, ShardValue<Comparable<?>> shardValue) {
        Collection<String> r = new HashSet<>();
        for (Object value : shardValue.getValues()) {
            String suffix = SEPARATOR + shardTarget.get( "" + rangeShardValue( (Date)value) );
            for (String shard : shardings) {
                if (shard.endsWith(suffix)) {
                    r.add(shard);
                }
            }
        }
        return r;
    }

    @Override
    public Collection<String> doBetween(Collection<String> shardings, ShardValue<Comparable<?>> shardValue) {
        throw new UnsupportedOperationException();
    }

    /**
     * 分片字段为 Date 类型
     * @param shardValue
     * @return
     */
    public int rangeShardValue(Date shardValue) {
        int i = 0;
        for (DateRange range : ranges) {
            if (range.inRange( shardValue.getTime() )) {
                return i;
            }
            i++;
        }
        throw new RuntimeException("No range found for " + shardValue);
    }

    /**
     * 读取文件中Range范围文本
     */
    private void readFile() {
        try {
            Properties rangeProperties = new Properties();
            rangeProperties.load( AbstractHashRangeShardAlgorithm.class.getClassLoader().getResourceAsStream( BASIC_DIR + fileName ) );

            String rangeText = rangeProperties.getProperty( "range" );
            if (null != rangeText && rangeText.length() > 0 && rangeText.indexOf( '-' ) > 0) {
                this.rangeText = rangeText;
            } else {
                throw new GenericException( RouteErrCodes.ROUTE_429, new Object[]{"读取时间分片路由算法配置文件:"+fileName+"错误"} );
            }
        } catch (IOException ioe) {
            logger.error( "Read hash range prop failed.", ioe );
            throw new GenericException( RouteErrCodes.ROUTE_428 );
        }
    }

    /**
     * 解析Range范围
     * 单字段分片时Range为一个List
     */
    private void parseRange() throws Exception{
        ranges = createRanges(rangeText);
    }

    protected List<DateRange> createRanges(String rangeContext) throws Exception{
        List<DateRange> ranges = new ArrayList<>();
        Iterable<String> rangeMetaArray = Splitter.on(",").trimResults().split(rangeContext);
        for (String rangeMeta : rangeMetaArray) {
            List<String> rangeItem = Splitter.on("-").trimResults().splitToList(rangeMeta);
            if (rangeItem.size() == 2) {
                ranges.add(new DateRange(rangeItem.get(0), rangeItem.get(1)));
            } else {
                logger.error("Invalid range configuration.", rangeText);
                throw new GenericException( RouteErrCodes.ROUTE_429, new Object[]{rangeMeta});
            }
        }
        return ranges;
    }

    /**
     * 时间分区对应的
     */
    protected void loadTargetMap() {
        int targetIndex = 0;
        for(int j=0; j<ranges.size(); j++){
            shardTarget.put(""+j, targetIndex++ +"");
        }
    }


    /**
     * 时间范围
     */
    public static class DateRange {
        private long start;
        private long end;

        DateRange(String startStr, String endStr) throws Exception{
            //String -- Date
            //在把一个字符串解析为日期的时候，请注意格式必须和给定的字符串格式匹配
            SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd HH:mm:ss");
            Date startDate = sdf.parse(startStr);
            this.start = startDate.getTime();
            Date endDate = sdf.parse(endStr);
            this.end = endDate.getTime();
        }

        public boolean inRange(long value) {
            return value >= start && value <= end;
        }
    }
}
